package com.ds.lens.flink.monitor.eventsinkjob;

import com.ds.lens.data.common.document.dto.MonitorEvent;
import com.ds.lens.flink.monitor.MonitorJobConstants;
import com.ds.lens.flink.monitor.MonitorJobContext;
import com.ds.lens.flink.monitor.common.CommonDeserializationSchema;
import com.ds.lens.flink.monitor.common.LensElasticsearchSink;
import com.ds.lens.flink.monitor.eventsinkjob.function.MonitorEventNotifyFunction;
import com.ds.lens.flink.monitor.eventsinkjob.function.MonitorEventSinkFunction;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.http.HttpHost;

import java.util.List;
import java.util.Map;
import java.util.Properties;

public class MonitorEventSinkJob {
    public static void main(String[] args) throws Exception {
        Properties properties = MonitorJobContext.getInstance().getKafkaProperties();
        if (properties == null) {
            System.err.println("No kafka properties found");
            return;
        }
        Properties kafkaProps = new Properties();
        kafkaProps.putAll(properties);
        kafkaProps.setProperty(MonitorJobConstants.KAFKA_CONSUMER_GROUP_CONFIG_FIELD, MonitorEventSinkJob.class.getName());

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);
        env.getConfig().setParallelism(MonitorJobContext.getInstance().getJobParallelism());

        Map<String, String> esConfig = MonitorJobContext.getInstance().getEsConfig();
        String consumeTopic = MonitorJobContext.getInstance().getMonitorEventTopic();
        List<HttpHost> transportAddresses = MonitorJobContext.getInstance().getEsTransportAddresses();
        env.addSource(new FlinkKafkaConsumer<>(consumeTopic,
                new CommonDeserializationSchema<>(MonitorEvent.class), kafkaProps))
                .process(new MonitorEventNotifyFunction())
                .addSink(new LensElasticsearchSink<>(transportAddresses, new MonitorEventSinkFunction(), esConfig).build());
        env.execute(MonitorEventSinkJob.class.getSimpleName());
    }
}
